package per.chao.mqtt.domain.protocol.unsubscribe;

import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import per.chao.mqtt.domain.protocol.BaseProtocolHandler;
import per.chao.mqtt.domain.repo.SubscribeMessageRepo;

import java.util.List;

/**
 * Description:
 *
 * @author W.Chao
 * @date 2021/6/19 23:05
 **/
@Component
@Slf4j
public class UnSubscribe extends BaseProtocolHandler {

	@Autowired
	private SubscribeMessageRepo subscribeMessageRepo;

	@Override
	public void handleMessage(Channel channel, Object msg) {
		MqttUnsubscribeMessage unsubscribeMessage = (MqttUnsubscribeMessage) msg;
		List<String> topics = unsubscribeMessage.payload().topics();
		String clientId = (String) channel.attr(AttributeKey.valueOf("clientId")).get();
		topics.forEach(topic -> {
			subscribeMessageRepo.remove(topic, clientId);
			log.info("UNSUBCRIBE - clientId: {}, topic: {}", clientId, topic);
		});
		MqttUnsubAckMessage unsubAckMessage = (MqttUnsubAckMessage) MqttMessageFactory.newMessage(
				new MqttFixedHeader(MqttMessageType.UNSUBACK, false, MqttQoS.AT_MOST_ONCE, false, 0),
				MqttMessageIdVariableHeader.from(unsubscribeMessage.variableHeader().messageId()),
				null);
		channel.writeAndFlush(unsubAckMessage);
	}
}
